package top.fullj.remoting.consumer;

import com.google.common.io.BaseEncoding;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import top.fullj.remoting.FutureValue;
import top.fullj.remoting.message.InvokeRequest;
import top.fullj.remoting.message.InvokeResponse;
import top.fullj.remoting.message.Message;

import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author bruce.wu
 * @since 2022/2/14 10:29
 */
public class Invocation {

    private static final Invocation INSTANCE = new Invocation();

    public static Invocation getInstance() {
        return INSTANCE;
    }

    private final IdGenerator idGenerator = new IdGenerator();

    public Object invoke(InvokeContext ctx) throws Throwable {
        Method method = ctx.getInvokeMethod();
        Class<?> c = method.getDeclaringClass();
        ServiceConfig config = getConfig(c.getAnnotation(Service.class));

        ctx.setServiceConfig(config);

        String id = idGenerator.next();
        AttributeKey<FutureValue<Message>> msgId = AttributeKey.valueOf(id);

        Channel channel = ConnectionPool.getInstance().borrow(config);
        try {
            InvokeRequest request = new InvokeRequest(config.getName(),
                    method.getName(), ctx.getInvokeArgs(), method.getParameterTypes());;

            Message message = new Message(request);
            message.setId(id);

            //TODO customized message interceptor chain before send

            FutureValue<Message> future = new FutureValue<>();
            channel.attr(msgId).set(future);

            //send request
            channel.writeAndFlush(message);

            //wait response
            Message reply;
            if (config.getTimeout() <= 0) {
                reply = future.get();
            } else {
                reply = future.get(config.getTimeout(), TimeUnit.MILLISECONDS);
            }

            //TODO customized message interceptor chain after receive

            //process result
            InvokeResponse response = (InvokeResponse) reply.getPayload();
            if (!response.isSuccess()) {
                throw new RemoteException(response.getMessage());
            }
            return response.getResult();
        } finally {
            channel.attr(msgId).set(null);
            ConnectionPool.getInstance().back(channel);
        }
    }

    private ServiceConfig getConfig(Service service) {
        return new ServiceConfig(service.name(), service.uri(), service.timeout());
    }

    static class IdGenerator {

        private final AtomicLong count = new AtomicLong(1);

        String next() {
            long number = count.getAndIncrement() & 0x7FFFFFFFFFFFFFFFL;
            ByteBuffer buf = ByteBuffer.allocate(8);
            buf.putLong(number);
            return BaseEncoding.base16().encode(buf.array());
        }

    }

}
